package com.supermarket.processors;

import com.supermarket.utils.JDBCUtil;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.streaming.api.java.JavaDStream;

import java.sql.Connection;
import java.sql.PreparedStatement;

public class ProductInOutCountProcessor {
    // 统计每种商品的上架 (U) 和下架 (D) 数量。
    public static void process(JavaDStream<String> jsonDStream, SparkSession spark) {
        jsonDStream.foreachRDD(rdd -> {
            if (rdd.isEmpty()) return;

            //将 JSON 数据解析为 DataFrame。
            //使用 SQL 查询按商品分组，计算 in_count（上架）和 out_count（下架）。
            //插入或更新 MySQL 表 rt_product_in_out_count
            Dataset<Row> df = spark.read().json(spark.createDataset(rdd.rdd(), Encoders.STRING()));
            df.createOrReplaceTempView("supermarket_data");

            Dataset<Row> result = spark.sql(
                    "SELECT " +
                            "  gid, gname, " +
                            "  SUM(CASE WHEN u_d = 'U' THEN u_d_num ELSE 0 END) as in_count, " +
                            "  SUM(CASE WHEN u_d = 'D' THEN u_d_num ELSE 0 END) as out_count " +
                            "FROM supermarket_data " +
                            "GROUP BY gid, gname"
            );

            result.foreach(row -> {
                try (Connection conn = JDBCUtil.getConnection();
                     PreparedStatement stmt = conn.prepareStatement(
                             "INSERT INTO rt_product_in_out_count (gid, gname, in_count, out_count) " +
                                     "VALUES (?, ?, ?, ?) " +
                                     "ON DUPLICATE KEY UPDATE " +
                                     "in_count = in_count + VALUES(in_count), " +
                                     "out_count = out_count + VALUES(out_count)")) {

                    stmt.setString(1, row.getString(0));
                    stmt.setString(2, row.getString(1));
                    stmt.setLong(3, row.getLong(2));
                    stmt.setLong(4, row.getLong(3));
                    stmt.executeUpdate();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
        });
    }
}